// -*- C++ -*-
//
//  This file is part of Siena, a wide-area event notification system.
//  See http://www.inf.usi.ch/carzaniga/siena/
//
//  Authors: Antonio Carzaniga
//  See the file AUTHORS for full details. 
//
//  Copyright (C) 2001-2005 University of Colorado
//
//  This program is free software; you can redistribute it and/or
//  modify it under the terms of the GNU General Public License
//  as published by the Free Software Foundation; either version 2
//  of the License, or (at your option) any later version.
//
//  This program is distributed in the hope that it will be useful,
//  but WITHOUT ANY WARRANTY; without even the implied warranty of
//  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//  GNU General Public License for more details.
//
//  You should have received a copy of the GNU General Public License
//  along with this program; if not, write to the Free Software
//  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307,
//  USA, or send email to one of the authors.
//
//
// $Id: fwd_table.cc,v 1.31 2010-03-11 15:16:40 carzanig Exp $
//
#include <new>

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include "siena/types.h"
#include "siena/exceptions.h"
#include "siena/stst.h"
#include "siena/forwarding.h"
#include "siena/allocator.h"
#include "siena/fwdtable.h"

#include "bitvector.h"
#ifndef WITH_STATIC_COUNTERS
#include "patricia.h"
#endif
#include "bloom_set.h"
#include "fwd_table.h"

#include "string_index.h"
#include "bool_index.h"
#include "constraint_index.h"
#include "mmhwrapper.h"

#ifdef WITH_STATISTICS
#include "statistics.h"
#endif

namespace siena {

/*  filter descriptor in FwdTable
 *
 *  Represents a conjunction of constraints in the data structures of
 *  the FwdTable implementation.
 */
class Filter {
public:
	/*  constructs a filter descriptor associating the filter with the
	 *  given interface
	 */
	Filter(Interface * xi, unsigned int xid) throw()
#ifdef WITH_STATIC_COUNTERS
	: i(xi), size(0), msg_id(0), count(0) {};
#else
	: i(xi), size(0), id(xid) {};
#endif

	/*  descriptor of the interface associated with this filter
	 *
	 *  The current implementation is not capable of recognizing two
	 *  identical filters.  Therefore, every filter will generate a
	 *  filter descriptor like this, and therefore a filter descriptor
	 *  is associated with a single interface.  This is why this is
	 *  simply a pointer to an interface descriptor.
	 *
	 *  <p>At some point, we migh want to be able to figure out that
	 *  F1==F2, and therefore to allow each filter to refer to more
	 *  than one interfaces.  In that case, we will have to maintain a
	 *  set of pointers to interface descriptors.
	 */
	Interface *			i;

	/*  number of constraints composing this filter
	 *
	 *  Notice that the implicit restriction of this declaration is
	 *  that we do not allow more than 255 constraints per filter.
	 */
	unsigned char		size;

#ifdef WITH_STATIC_COUNTERS
	/*  message id
	 */
	mutable unsigned int	msg_id;

	/*  number of constraints matched so far for this message id
	 */
	mutable unsigned char	count;
#else
	/*  filter identifier used as a key in the table of counters in
	 *  the matching algorithm
	 */
	unsigned int		id;
#endif
	/*  Bloom signature for this filter
	 */
	bloom_set			bset;
};

/* attribute descriptor
 *
 *  this class represents a constraint name within the constraint
 *  index.  It holds all the type-specific indexes for the constraints
 *  pertaining to this attribute.
 */
class AttributeType {
public:
	/* set of interfaces for which this attribute is a
	 *  <em>determinant</em> attribute.
	 *
	 *  AttributeType <em>A</em> is <em>determinant</em> for interface
	 *  <em>I</em> if all the conjuncts (i.e., filters) contained in
	 *  the disjunct (i.e., predicate) associated with <em>I</em>
	 *  contain at least one constraint of <em>A</em>.
	 */
	Selectivity *		exclude;

	/* index of <em>anytype</em> constraints */
	ConstraintType *		any_value_any_type;
	/* index of <em>integer</em> constraints */
	constraint_index<int_t>	int_index;
	/* index of <em>double</em> constraints */
	constraint_index<double_t>	double_index;
	/* index of <em>string</em> constraints */
	StringIndex			str_index;
	/* index of <em>boolean</em> constraints */
	BoolIndex			bool_index;
	//
	// more constraints can be added here (for example, for Time)
	// ...to be continued...
	//

	/*  adds the given constraints to the appropriate constraint
	 *	index associated with this attribute name.
	 */
	ConstraintType *		add_constraint(const constraint &,
			FTAllocator &)
			throw(bad_constraint);
	/*  constructor */
	AttributeType() throw()
			: exclude(0), any_value_any_type(0),
			  int_index(), double_index(), str_index(), bool_index() {};

	void			consolidate(FTAllocator & ftm) throw();
};

void AttributeType::consolidate(FTAllocator & ftm) throw() {
	int_index.consolidate(ftm);
	double_index.consolidate(ftm);
	str_index.consolidate(ftm);
}

/** a selectivity descriptor
 *
 *  associates a <em>selectivity</em> to a given attribute/constraint
 *  name.  The selectivity associated with a name is given by a set of
 *  interfaces that can not be matched by a message unless that
 *  attribute is present in the message.
 **/
class Selectivity {
public:
	/** creates a selectivity item with the given name, excluding the
	 *	given interface and connecting it to the given previous
	 *  element in the selectivity table.
	 **/
	Selectivity(const char * s, const char * t, ifid_t i, Selectivity * prev)
	throw()
	: name(s,t), exclude(), prev(prev), next(0) {}

	/** constraint/attribute name **/
	string_t			name;
	/** set of interfaces excluded by the absence of this attribute **/
	ibitvector			exclude;

	/** pointer to the previous element in the selectivity table.
	 *
	 *  Points to an element with greater or equal selectivity
	 **/
	Selectivity *		prev;

	/** pointer to the next element in the selectivity table.
	 *
	 *  Points to an element with less or equal selectivity
	 **/
	Selectivity *		next;
};

/** Interface descriptor.
 *
 *  Stores the association between user-supplied interface identifier
 *  and interface identifier used internally by the forwarding table.
 *  The reason for maintaining two identifiers is that identifiers
 *  used by the matching and pre-processing functions must be
 *  contiguous.  And we don't want to rely on the user to supply
 *  contiguous identifiers.
 **/
class Interface {
public:
	/** user-defined interface identifier **/
	const if_t			interface;

	/** internal identifier used by the matching function **/
	const ifid_t		id;

	/** builds an interface descriptor with the given identifiers **/
	Interface(if_t xif, ifid_t xid) throw() : interface(xif), id(xid) {};
};

FwdTable::FwdTable() throw()
    		: if_count(0), f_count(0), preprocess_rounds(10),
    		  attributes(), selectivity_first(0), selectivity_last(0),
    		  consolidate_list(0) { }

FwdTable::~FwdTable() throw() {
	consolidate();
	clear();
}

void FwdTable::clear() throw() {
	memory.clear();
	attributes.clear();
	if_count = 0;
	selectivity_first = 0;
	selectivity_last = 0;

	attr_descr_link * tmp;
	while(consolidate_list) {
		tmp = consolidate_list;
		consolidate_list = consolidate_list->next;
		delete(tmp);
	}
}

void FwdTable::clear_recycle() throw() {
	memory.recycle();
	attributes.clear();
	if_count = 0;
	selectivity_first = 0;
	selectivity_last = 0;

	attr_descr_link * tmp;
	while(consolidate_list) {
		tmp = consolidate_list;
		consolidate_list = consolidate_list->next;
		delete(tmp);
	}
}

size_t FwdTable::bytesize() const throw() {
	return memory.size();
}

size_t FwdTable::allocated_bytesize() const throw() {
	return memory.allocated_size();
}

Selectivity * 
FwdTable::new_selectivity(const string_t & name, ifid_t i) throw() {
	char * s = new (memory) char[name.length()];
	char * t = s;
	const char * c = name.begin;
	while (c != name.end) *t++ = *c++;

	selectivity_last = new (memory) Selectivity(s, t, i, selectivity_last);


	selectivity_last->exclude.set(i, memory);

	if (selectivity_first == 0)
		selectivity_first = selectivity_last;
	return selectivity_last;
}

void FwdTable::add_to_selectivity(Selectivity * s, ifid_t i) throw() {
	s->exclude.set(i, memory);	// this increases the selectivity level by 1
	Selectivity * sp = s->prev;
	if (sp == 0) return;

	while (sp->exclude.get_count() < s->exclude.get_count()) {
		// as long as the predecessor of s has
		sp->next = s->next;	// a lower selectivity level,
		s->next = sp;		// swap s and its predecessor
		if (sp->next == 0)
			selectivity_last = sp;

		s->prev = sp->prev;
		sp->prev = s;
		if (s->prev == 0) {
			selectivity_first = s;
			return;
		}
		sp = s->prev;
	}
}

AttributeType * FwdTable::get_attribute(const string_t & name) throw() {
	STSTNode * np = attributes.insert(name.begin, name.end, memory);
	if (np->data == 0) {
		np->data = new (memory) AttributeType();
		//
		// we add every new attribute descriptor to consolidate_list
		//
		attr_descr_link * ad_link = new attr_descr_link;
		ad_link->a = static_cast<AttributeType*>(np->data);
		ad_link->next = consolidate_list;
		consolidate_list = ad_link;
	}

	return static_cast<AttributeType*>(np->data);
}

void FwdTable::consolidate() throw() {
	attr_descr_link * tmp;
#ifdef WITH_STATISTICS
	sff_timer_stack.push(consolidate_timer);
#endif
	while(consolidate_list) {
		consolidate_list->a->consolidate(memory);
		tmp = consolidate_list;
		consolidate_list = consolidate_list->next;
		delete(tmp);
	}
#ifdef WITH_STATISTICS
	sff_timer_stack.pop();
#endif
}

ConstraintType * AttributeType::add_constraint(const constraint & c,
		FTAllocator & a) throw(bad_constraint) {
	switch (c.type()) {
	case anytype_id:
		switch (c.op()) {
		case any_id:
			if (any_value_any_type == 0)
				any_value_any_type =  new (a) ConstraintType();
			return any_value_any_type;
		default:
			throw(bad_constraint("bad operator for anytype."));
		}
		case int_id:
			switch (c.op()) {
			case any_id: return int_index.add_any(a);
			case eq_id: return int_index.add_eq(c.int_value(), a);
			case gt_id: return int_index.add_gt(c.int_value(), a);
			case lt_id: return int_index.add_lt(c.int_value(), a);
			case ne_id: return int_index.add_ne(c.int_value(), a);
			default:
				throw(bad_constraint("bad operator for int."));
			}
			break;
			case string_id:
				switch (c.op()) {
				case any_id: return str_index.add_any(a);
				case eq_id: return str_index.add_eq(c.string_value(), a);
				case gt_id: return str_index.add_gt(c.string_value(), a);
				case lt_id: return str_index.add_lt(c.string_value(), a);
				case pf_id: return str_index.add_pf(c.string_value(), a);
				case sf_id: return str_index.add_sf(c.string_value(), a);
				case ss_id: return str_index.add_ss(c.string_value(), a);
				case re_id: return str_index.add_re(c.string_value(), a);
				case ne_id: return str_index.add_ne(c.string_value(), a);
				default:
					throw(bad_constraint("bad operator for string."));
				}
				break;
				case bool_id:
					switch (c.op()) {
					case any_id: return bool_index.add_any(a);
					case eq_id: return bool_index.add_eq(c.bool_value(), a);
					case ne_id: return bool_index.add_ne(c.bool_value(), a);
					default:
						throw(bad_constraint("bad operator for boolean."));
					}
					break;
					case double_id:
						switch (c.op()) {
						case any_id: return double_index.add_any(a);
						case eq_id: return double_index.add_eq(c.double_value(), a);
						case gt_id: return double_index.add_gt(c.double_value(), a);
						case lt_id: return double_index.add_lt(c.double_value(), a);
						case ne_id: return double_index.add_ne(c.double_value(), a);
						default:
							throw(bad_constraint("bad operator for double."));
						}
						break;
						default:
							throw(bad_constraint("bad type."));
	}
	return 0;
}

void 
FwdTable::connect(ConstraintType * c, Filter * f, const string_t & name) throw() {
	if (c->f == 0) {
		c->f = f;
	} else {
		c->next = new (memory)flist(f, c->next);
	}
	++(f->size);
	f->bset.add_string(name.begin, name.end);
}

// element of the table of candidate selective attributes used in
// ifconfig (below) to compute the selective attrs of a predicate
//
struct s_table_elem {
	string_t name;
	AttributeType * attr;
	s_table_elem * next;

	s_table_elem(const string_t & n, AttributeType * a, s_table_elem * x) throw()
			: name(n), attr(a), next(x) {};
};

//  This is the main method that builds the forwarding table.  It is
//  conceptually simple, but it requires a bit of attention to manage
//  the compilation of the selectivity table.
//
void FwdTable::ifconfig(if_t ifx, const predicate & p) 
throw (bad_constraint, std::bad_alloc) {
	predicate::iterator * pi = p.first();
	if (!pi) return;

#ifdef WITH_STATISTICS
	sff_timer_stack.push(ifconfig_timer);
#endif

	tmp_memory.recycle();	// we'll use some tmp mem, so we
	// recycle it first
	Interface * i;
	Filter * f;
	ConstraintType * c;

	s_table_elem * s_attrs = 0;	// list of selective attributes
	s_table_elem * si; 		// iterator for s_attrs

	AttributeType *		adescr;

	i = new (memory) Interface(ifx, if_count++);
	f = 0;			// now *pi is the first filter of
	filter::iterator * fi;	// predicate p, therefore we simply
	if ((fi = pi->first())) {	// fill s_attrs with every constraint
		do {			// name in *pi
			string_t name = fi->name();
			adescr = get_attribute(name);
			s_attrs = new (tmp_memory) s_table_elem(name, adescr, s_attrs);

			c = adescr->add_constraint(*fi, memory);
			if (c != 0) {	// here we build the forwarding table
				if (f == 0)	// connecting constraints to the filter
					f = new (memory) Filter(i, ++f_count);
				connect(c, f, name);
			}
		} while (fi->next());
		delete(fi);
	}				// then for every following filter *pi
	while(pi->next()) {		// in p we intersect s_attrs with the
		f = 0;			// sets of attributes of each of *pi
		if ((fi = pi->first())) {
			s_table_elem * intersection = 0;
			do {
				string_t name = fi->name();
				c = 0;

				s_table_elem ** dprev = &s_attrs;
				while((si = *dprev)) {
					if (si->name == name) {     // find name in d_attr list
						*dprev = si->next;	// move si from d_attr list
						si->next = intersection;// to intersection list
						intersection = si;
						// and since we found the attribute, we directly use
						// the attribute descriptor stored in the list
						c = si->attr->add_constraint(*fi, memory);
						break;
					} else {
						dprev = &(si->next);
					}
				}
				if (c == 0)
					c = get_attribute(name)->add_constraint(*fi, memory);
				if (c != 0) {
					if (f == 0)
						f = new (memory) Filter(i, ++f_count);
					connect(c, f, name);
				}
			} while (fi->next());
			delete(fi);
			s_attrs = intersection;
		}
	}
	delete(pi);		  // we then add this interface to the exclude
	while(s_attrs) {	  // set of each of the selective attributes
		if (s_attrs->attr->exclude) {
			add_to_selectivity(s_attrs->attr->exclude, i->id);
		} else {
			s_attrs->attr->exclude = new_selectivity(s_attrs->name, i->id);
		}
		s_attrs = s_attrs->next;
	}
#ifdef WITH_STATISTICS
	sff_timer_stack.pop();
#endif
}

void FwdTable::set_preprocess_rounds(unsigned int i) throw() {
	preprocess_rounds = i;
}

unsigned int FwdTable::get_preprocess_rounds() const throw() {
	return preprocess_rounds;
}

void 
FwdTable::match(const message & m, MatchMessageHandler & p) const throw() {
	MMHWrapper w(p, m);
	match(m, w);
}

void FwdTable::match(const message & n, MatchHandler & p) const throw() {
	const STSTNode * tn;
	const AttributeType * adescr;

	message::iterator * i = n.first ();
	if (!i)
		return;

#ifdef WITH_STATISTICS
	sff_timer_stack.push(match_timer);
#endif

	//
	// first we construct a constraint matcher object
	//
	bitvector mask(if_count);
	//
	// pre-processing
	//
	const Selectivity * si = selectivity_first;
	unsigned int rounds = preprocess_rounds;
	while (rounds > 0 && si != 0 && mask.get_count() < if_count) {
		if (!n.contains(si->name))
			mask.set(si->exclude);
		si = si->next;
		--rounds;
	}
	if (mask.get_count() >= if_count) {
#ifdef WITH_STATISTICS
		sff_timer_stack.pop();
#endif
		return;
	}
	//
	//  here we go, the outer loop extracts the set of matching
	//  constraints for each attribute in n
	//
	CProcessor cp(if_count, p, &mask, n);
	do {
		string_t name = i->name();
		tn = attributes.find(name.begin, name.end);
		if (tn != 0) {
			adescr = static_cast<AttributeType *>(tn->data);
			if (adescr->any_value_any_type)
				if (cp.process_constraint(adescr->any_value_any_type))
					goto cleanup_and_return;
			switch(i->type()) {
			case int_id:
				if (adescr->int_index.match(i->int_value(), cp))
					goto cleanup_and_return;
				if (adescr->double_index.match(i->int_value(), cp))
					goto cleanup_and_return;
				break;
			case string_id:
				if (adescr->str_index.match(i->string_value(), cp))
					goto cleanup_and_return;
				break;
			case bool_id:
				if (adescr->bool_index.match(i->bool_value(), cp))
					goto cleanup_and_return;
				break;
			case double_id:
				if (adescr->double_index.match(i->double_value(), cp))
					goto cleanup_and_return;
				if (adescr->int_index.match((int)i->double_value(), cp))
					goto cleanup_and_return;
				break;
			default:
				// Unknown type: this is a malformed attribute.  In
				// this case we simply move along with the matching
				// function.  Obviously, we could (1) throw an
				// exception, (2) print an error message, or (3) ask
				// the user (i.e., the programmer) to supply an error
				// callback function and call that function.  This is
				// largely a semantic issue.  Notice that returning
				// immediately (from the matching function) would not
				// be a good idea, since we might have already matched
				// some predicates, and therefore we might end up
				// excluding some other predicates based only on the
				// ordering of attributes.
				//
				break;
			}
			if (mask.get_count() >= if_count) goto cleanup_and_return;
		}
	} while (i->next());
	cleanup_and_return:
	delete(i);
#ifdef WITH_STATISTICS
	sff_timer_stack.pop();
#endif
}

bool CProcessor::process_constraint(const ConstraintType * c) throw() {
	//
	// we look at every filter in which the constraint appears
	//
	const Filter * f = c->f;
	const flist * k = c->next;
	for(;;) {
		if (bset.covers(f->bset)) {
#ifdef WITH_STATIC_COUNTERS
			if (f->msg_id == msg_id) {
				++f->count;
			} else {
				f->msg_id = msg_id;
				f->count = 1;
			}
			if (f->count >= f->size) {
				if (! if_mask->test(f->i->id)) {
#else
					if (! if_mask->test(f->i->id)) {
						if (++fmap[f->id] >= f->size) {
#endif
							if_mask->set(f->i->id);
#ifdef WITH_STATISTICS
							sff_timer_stack.push(forward_timer);
#endif
							bool output_result = processor.output(f->i->interface);
#ifdef WITH_STATISTICS
							sff_timer_stack.pop();
#endif
							if (output_result)
								return true;
							if (if_mask->get_count() >= target)
								return true;
							//
							// we immediately return true if the processing
							// function returns true or if we matched all
							// possible interfaces
							//
						}
					}
				}
				//
				// here's where I could do something intelligent
				// about other filters pointing to the same
				// interface: (*k)->interface->filters
				//
				if (k == 0) break;
				f = k->f;
				k = k->next;
			}

			return false;
		}

	} // namespace siena
